
package com.zxq.gmall.realtime.util

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.{RequestOptions, RestClient,
  RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType
/**
  * ES 工具类
  */
object MyEsUtils {
  //声明 es 客户端
  var esClient : RestHighLevelClient = build()

  /**
    * 批量数据幂等写入
    * 通过指定 id 实现幂等
    */
  def bulkSaveIdempotent(sourceList: List[(String , AnyRef)] ,indexName: String ): Unit ={
    if(sourceList!= null && sourceList.nonEmpty){
      val bulkRequest = new BulkRequest()
      for ((docId,sourceObj) <- sourceList) {
        val indexRequest = new IndexRequest()
        indexRequest.index(indexName)
        val movieJsonStr: String = JSON.toJSONString(sourceObj,
          new SerializeConfig(true))
        indexRequest.source(movieJsonStr,XContentType.JSON)
        indexRequest.id(docId)
        bulkRequest.add(indexRequest)
      }
      esClient.bulk(bulkRequest,RequestOptions.DEFAULT)
    }
  }
/*** 批量数据写入
  */
def bulkSave(sourceList: List[AnyRef] ,indexName: String ):
Unit ={
  // BulkRequest 实际上就是由多个单条 IndexRequest 的组合
  val bulkRequest = new BulkRequest()
  for (source <- sourceList) {
    val indexRequest = new IndexRequest()
    indexRequest.index(indexName)
    val movieJsonStr: String = JSON.toJSONString(source, new
        SerializeConfig(true))
    indexRequest.source(movieJsonStr,XContentType.JSON)
    bulkRequest.add(indexRequest)
  }
  esClient.bulk(bulkRequest,RequestOptions.DEFAULT)
}
  /**
    * 单条数据幂等写入
    * 通过指定 id 实现幂等
    */
  def saveIdempotent(source:(String , AnyRef),indexName :
  String ): Unit = {
    val indexRequest = new IndexRequest()
    indexRequest.index(indexName)
    val movieJsonStr: String = JSON.toJSONString(source._2, new
        SerializeConfig(true))
    indexRequest.source(movieJsonStr,XContentType.JSON)
    indexRequest.id(source._1)
    esClient.index(indexRequest,RequestOptions.DEFAULT)
  }
  /**
    * 单条数据写入
    */
  def save(source:AnyRef,indexName : String ): Unit = {
    val indexRequest = new IndexRequest()
    indexRequest.index(indexName)
    val movieJsonStr: String = JSON.toJSONString(source, new
        SerializeConfig(true))
    indexRequest.source(movieJsonStr,XContentType.JSON)
    esClient.index(indexRequest,RequestOptions.DEFAULT)
  }
  /**
    * 销毁
    */
  def destory(): Unit ={
    esClient.close()
    esClient = null
  }
/**
  * 创建 es 客户端对象
  */def build():RestHighLevelClient = {
  val builder: RestClientBuilder = RestClient.builder(new HttpHost("hadoop102",9200))
  val esClient = new RestHighLevelClient(builder)
  esClient
}
  /**
    * 获取 esclient
    */
  def getClient(): RestHighLevelClient = {
    esClient
  }
}

